package com.exd.flink.stream.function;

import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;

import com.exd.flink.common.DomainControlFlag;
import com.exd.flink.common.OutputVdataTagBase;

/**
 * 描述: 进行数据的分流
 * 创建人: 赵兴炎
 * 日期: 2022年9月13日
 */
public class ShuntStreamFunction extends ProcessAllWindowFunction<String,String,GlobalWindow>{

	private static final long serialVersionUID = 1L;
	
	@Override
	public void process(Context context, Iterable<String> elements, Collector<String> collector) throws Exception {
		elements.forEach(data -> {
			// 解析部分字符串进行数据的分类
			String dataPartValue = data.substring(0, 200);
			// 根据不同的域控进行数据分流
			if(dataPartValue.contains(":"+DomainControlFlag.BDCM_VDATA))  context.output(OutputVdataTagBase.BDCM_VDATA, data);
			if(dataPartValue.contains(":"+DomainControlFlag.VDCM_VDATA))  context.output(OutputVdataTagBase.VDCM_VDATA, data);
			if(dataPartValue.contains(":"+DomainControlFlag.BDCM_VCOMPUTE))  context.output(OutputVdataTagBase.BDCM_VCOMPUTE, data);
			if(dataPartValue.contains(":"+DomainControlFlag.VDCM_VCOMPUTE))  context.output(OutputVdataTagBase.VDCM_VCOMPUTE, data);
		});
	}

}
